package com.yht.redbook.note.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.yht.framework.common.utils.JsonUtils;
import com.yht.redbook.note.biz.constant.MQConstants;
import com.yht.redbook.note.biz.domain.dataobject.NoteLikeDO;
import com.yht.redbook.note.biz.domain.mapper.NoteLikeDOMapper;
import com.yht.redbook.note.biz.model.dto.LikeUnlikeNoteMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

@RocketMQMessageListener(consumerGroup = "redbook_group_" + MQConstants.TOPIC_LIKE_OR_UNLIKE,
        topic = MQConstants.TOPIC_LIKE_OR_UNLIKE, consumeMode = ConsumeMode.ORDERLY)
@Component
@Slf4j
public class LikeUnlikeNoteConsumer implements RocketMQListener<Message> {
    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private NoteLikeDOMapper noteLikeDOMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();

        // 幂等性：通过联合唯一索引保证

        // 消息体
        String body = new String(message.getBody());

        // 标签
        String tags = message.getTags();

        log.info("==> LikeUnlikeNoteConsumer 消费了消息 {}，tags: {}", body, tags);

        if (Objects.equals(MQConstants.TAG_LIKE, tags)) {
            handleLikeNoteTagMessage(body);
        } else if (Objects.equals(MQConstants.TAG_UNLIKE, tags)) {
            handleUnLikeNoteTagMessage(body);
        }
    }

    /**
     * 笔记取消点赞
     *
     * @param body
     */
    private void handleUnLikeNoteTagMessage(String body) {
        LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO = JsonUtils.parseObject(body, LikeUnlikeNoteMqDTO.class);

        if (Objects.isNull(likeUnlikeNoteMqDTO)) return;

        Long userId = likeUnlikeNoteMqDTO.getUserId();
        Long noteId = likeUnlikeNoteMqDTO.getNoteId();
        Integer type = likeUnlikeNoteMqDTO.getType();
        LocalDateTime createTime = likeUnlikeNoteMqDTO.getCreateTime();

        NoteLikeDO noteLikeDO = NoteLikeDO.builder()
                .noteId(noteId)
                .userId(userId)
                .createTime(createTime)
                .status(type).build();

        int count = noteLikeDOMapper.update2UnlikeByUserIdAndNoteId(noteLikeDO);
        // 发送计数 MQ
        if (count == 0) return;

        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(body).build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【笔记取消点赞】 MQ 发送成功，SendResult：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【笔记取消点赞】 MQ 发送异常，", throwable);
            }
        });
    }

    /**
     * 笔记点赞
     *
     * @param body
     */
    private void handleLikeNoteTagMessage(String body) {
        LikeUnlikeNoteMqDTO likeUnlikeNoteMqDTO = JsonUtils.parseObject(body, LikeUnlikeNoteMqDTO.class);

        if (Objects.isNull(likeUnlikeNoteMqDTO)) return;

        Long userId = likeUnlikeNoteMqDTO.getUserId();
        Long noteId = likeUnlikeNoteMqDTO.getNoteId();
        Integer type = likeUnlikeNoteMqDTO.getType();
        LocalDateTime createTime = likeUnlikeNoteMqDTO.getCreateTime();

        NoteLikeDO noteLikeDO = NoteLikeDO.builder()
                .id(noteId)
                .userId(userId)
                .status(type)
                .createTime(createTime)
                .build();

        int count = noteLikeDOMapper.insertOrUpdate(noteLikeDO);

        // 发送计数 MQ
        if (count == 0) return;

        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(body).build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【笔记点赞】 MQ 发送成功，SendResult：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【笔记点赞】 MQ 发送异常，", throwable);
            }
        });
    }
}
